package com.zzy.chapter05.trans;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransTupleAggragationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Tuple3<String, String, Integer>> source = env.fromElements(
                Tuple3.of("a", "1", 5),
                Tuple3.of("a", "2", 3),
                Tuple3.of("b", "3", 3),
                Tuple3.of("b", "4", 4)
        );

        //source.keyBy(r -> r.f0).sum(1).print();
        //source.keyBy(r -> r.f0).sum("f1").print();

        //source.keyBy(r -> r.f0).max(1).print();
        //source.keyBy(r -> r.f0).max("f1").print();

        //source.keyBy(r -> r.f0).min(1).print();
        //source.keyBy(r -> r.f0).min("f2").print();//只输出最小的那个字段，其他字段不一定属于本数据

        //source.keyBy(r -> r.f0).maxBy(1).print();
        source.keyBy(r -> r.f0).minBy("f2").print(); //全部字段都是本数据的

        env.execute();
    }
}
